package sample.batch.paritition;

import java.util.Random;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.partition.support.SimplePartitioner;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;

/**
 * @date 2020/8/18
 */
@SpringBootApplication
@EnableBatchProcessing
public class PartitionApplication {
	private static final Logger LOGGER = LoggerFactory.getLogger(PartitionApplication.class);

	public static void main(String[] args) {
		SpringApplication.run(PartitionApplication.class, args);
	}

	@Autowired
	public JobBuilderFactory jobBuilderFactory;

	@Autowired
	public StepBuilderFactory stepBuilderFactory;

	@Bean
	public TaskExecutor taskExecutor() {
		SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
		taskExecutor.setConcurrencyLimit(3);
		return taskExecutor;
	}

	@Bean
	public Step step1() {
		return stepBuilderFactory.get("step1")
				.tasklet(new Tasklet() {
					@Override
					public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
//						if (new Random().nextBoolean()) {
//							throw new RuntimeException("mock");
//						}
						LOGGER.info("xxx");
						return RepeatStatus.FINISHED;
					}
				})
				.build();
	}

	@Bean
	public Step partitionStep() {
		return stepBuilderFactory.get("step1.manager")
				.partitioner("step1", new SimplePartitioner())
				.step(step1())
				.taskExecutor(taskExecutor())
				.gridSize(3)
				.build();
	}

	@Bean
	public Job partitionJob(@Qualifier("partitionStep") Step partitionStep) {
		return jobBuilderFactory.get("partitionJob")
				.start(partitionStep)
				.build();
	}

}
